package net.bwie.vehicle.dws.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ibm.icu.impl.Row;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import net.bwie.vehicle.dws.bean.CarBean;
import net.bwie.vehicle.dws.bean.TripDetectionResult;
import net.bwie.vehicle.dws.function.JourneyRealtimeFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/**
 * 行程实时分析任务
 */
public class JourneyRealtimeAnalyseJob {
    public static void main(String[] args) throws Exception {
        // 创建执行环境并设置并行度为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从Kafka消费数据
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "Realtime-vehicle-data");

            // 简化的处理流水线
        DataStream<TripDetectionResult> tripStream = kafkaStream
                // 解析JSON并处理数据
                .map(value -> {
                    try {
                        System.out.println("Processing: " + value); // 调试输出
                        return JSON.parseObject(value, CarBean.class);
                    } catch (Exception e) {
                        System.err.println("JSON parse error: " + value);
                        return null;
                    }
                })
                // 过滤掉解析失败的数据
                .filter(bean -> bean != null)
                // 按照车辆识别码分组
                .keyBy(CarBean::getVin)
                // 使用自定义函数处理流数据
                .process(new JourneyRealtimeFunction())
                // 设置任务名称
                .name("simplified-trip-detector");

                // 输出结果
        SingleOutputStreamOperator<String> map = tripStream.map(new MapFunction<TripDetectionResult, String>() {
            @Override
            public String map(TripDetectionResult value) throws Exception {
                // 构造输出字符串
                String output = value.getVin() + ',' +
                        value.getTripId() + ',' +
                        value.getStartTime() + ',' +
                        value.getEndTime() + ',' +
                        value.getDistance();
                System.out.println(output);
                return output;
            }

        });

// TODO: 2025/5/22 插入数据
        String upsertSql = "INSERT INTO table new_car.car_realtime (Vin, tripId, startTime, endTime,distance)\n" +
                "VALUES (?, ?, ?, ?,?)";
// TODO: 2025/5/22 插入数据
        JdbcUtil.sinkToClickhouseUpsert(map, upsertSql);

// 执行环境
        env.execute("xxxx");

    }
}